package cn.itcast.flink.examples;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;

/**
 * Author itcast
 * Date 2022/1/18 10:31
 * Desc 实现一个 wordcount 通过 FlinkTable&SQL
 * 输入表样式:
 * | hadoop | 1
 * | hello  | 1
 * | spark  | 2
 * select word,count(1)
 * from t_wordcount
 * group by word;
 */
public class WordcountTableDemo {
    public static void main(String[] args) throws Exception {
        //1.准备环境 获取流执行环境 流表环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //设置执行环境配置
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .useBlinkPlanner()
                .inStreamingMode()
                .build();
        //设置流表环境
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, settings);
        //2.Source 获取 单词信息
        DataStream<WC> input = env.fromElements(
                new WC("Hello", 1l),
                new WC("World", 1l),
                new WC("Hello",2l),
                new WC("FLINK",2l)
        );
        //3.创建视图 WordCount
        tEnv.createTemporaryView("t_wordcount",input,$("word"),$("frequency"));
        //4.从临时视图中获取一个 Table 对象
        //获取表对象的两种方法 ① from("t_wordcount")
        Table wordcount = tEnv.from("t_wordcount");
        // ② fromDataStream(ds,字段列表) 字段可以不加，推荐加上
        Table table1 = tEnv.fromDataStream(input, $("word"), $("frequency"));
        // 5.使用 DSL 语句实现 wordcount 单词统计
        Table result = wordcount.groupBy($("word"))
                .select(
                        $("word"),
                        $("word").count().as("cnt")
                );
        //将 wordcount table 转换成 DataStream
        DataStream<Tuple2<Boolean, Row>> rowDataStream = tEnv.toRetractStream(result, Row.class);
        //6.打印输出结果
        rowDataStream.print();
        //7.执行
        env.execute();
    }

    public static class WC {
        public String word;
        public Long frequency;
        public WC(){}

        public WC(String word, Long frequency) {
            this.word = word;
            this.frequency = frequency;
        }

        public String getWord() {
            return word;
        }

        public void setWord(String word) {
            this.word = word;
        }

        public Long getFrequency() {
            return frequency;
        }

        public void setFrequency(Long frequency) {
            this.frequency = frequency;
        }
    }
}
